package flink.serialize;

import org.apache.flink.api.common.serialization.SerializationSchema;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.apache.commons.lang3.StringUtils;

import javax.annotation.Nullable;

import java.util.Arrays;

/**
 * @author ：Jason
 * @date ：Created in 2023/5/5 12:02 PM
 * @description： 写入 Kafka 时 value 的序列化实现,在写入之前可以删除传入的字段
 * @modified By：Jason
 * @version: 1.0
 */
public class KafkaValueSerializeSchema implements SerializationSchema<JSONObject> {
    private final String SPLIT_INDEX = ",";
    private final @Nullable String removeKey;

    public KafkaValueSerializeSchema(@Nullable String removeKey) {
        this.removeKey = removeKey;
    }

    @Override
    public byte[] serialize(JSONObject element) {
        assert StringUtils.isNotEmpty(this.removeKey);
        Arrays.stream(this.removeKey.split(SPLIT_INDEX)).forEach(element::remove);
        return element.toString(SerializerFeature.WriteMapNullValue).getBytes();
    }
}
